package elephant.rpc.client;

import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import elephant.rpc.client.core.NetworkService;
import elephant.rpc.client.core.RPCAsyncInvocationHandler;
import elephant.rpc.client.core.RPCInvocationHandler;
import elephant.rpc.client.core.RPCMessageCallback;
import elephant.rpc.client.core.RPCServerInfo;
import elephant.rpc.client.core.RPCSyncInvocationHandler;
import elephant.rpc.client.core.ResponseLock;
import elephant.rpc.network.netty.NettyClientService;
import elephant.rpc.server.core.PerformMonitor;
import elephant.rpc.server.session.RPCSession;
import elephant.rpc.threadpool.ThreadPoolManager;
import elephant.utils.ObjectPrinter;

/**
 * 
 * @author skydu
 *
 */
public class RPCClient {
	//
	private static Logger logger=LoggerFactory.getLogger(RPCClient.class);
	//
	public static final String CODEC_TYPE_FASTJSON="fastjson";
	public static final String CODEC_TYPE_JDK="jdk";
	//
	public static String codecType=CODEC_TYPE_FASTJSON;
	//
	private String uuid;
	private ThreadPoolManager threadPoolManager;
	private Map<String,List<RPCServerInfo>> serverInfoMap;
	private NetworkService networkService;
	private PerformMonitor performMonitor;
	//
	public AtomicLong messageId;
	/**session list key:serverCluster*/
	private Map<String,List<RPCSession>>sessionMap;
	/**topic map*/
	private Map<String,Set<String>>topicMap;
	/**serivice map*/
	private Map<String,Object> syncServiceMap;
	/**async serivice map*/
	private Map<String,Object> asyncServiceMap;
	/**invocation handler**/
	private Map<String,RPCInvocationHandler> syncInvocationHandlerMap;
	/**invocation async handler**/
	private Map<String,RPCInvocationHandler> asyncInvocationHandlerMap;
	/**lock map*/
	private Map<Long, ResponseLock> syncLockMap;
	/**pushMessageCallback*/
	private RPCMessageCallback pushMessageCallback;
	/**rpc超时时间(20秒超时)*/
	private int rpcTimeout=20000;
	/**contextClassLoader*/
	private ClassLoader contextClassLoader;
	//
	public RPCClient(){
		this.uuid=UUID.randomUUID().toString();
		this.performMonitor=new PerformMonitor();
		this.threadPoolManager=new ThreadPoolManager(contextClassLoader);
		this.networkService=new NettyClientService(this);
		this.sessionMap=new ConcurrentHashMap<>();
		this.serverInfoMap=new ConcurrentHashMap<>();
		this.topicMap=new ConcurrentHashMap<>();
		this.messageId=new AtomicLong(1);
		this.syncServiceMap=new HashMap<>();
		this.asyncServiceMap=new HashMap<>();
		this.syncInvocationHandlerMap=new HashMap<>();
		this.asyncInvocationHandlerMap=new HashMap<>();
		this.syncLockMap = new ConcurrentHashMap<>();
	}
	//
	public void init() throws Exception {
		if(logger.isInfoEnabled()){
			logger.info(getClass().getSimpleName()+" init");
		}
		threadPoolManager.init();
		serverInfoMap.forEach((cluster,serverList)->{
			serverList.forEach(serverInfo->initSessionList(serverInfo));
		});
		networkService.init();
	}
	//
	private void initSessionList(RPCServerInfo serverInfo){
		RPCSession session=new RPCSession();
		session.clientUuid=uuid;
		session.remoteHost=serverInfo.host;
		session.remotePort=serverInfo.port;
		session.serverCluster=serverInfo.cluster;
		List<RPCSession>sessionList=sessionMap.get(serverInfo.cluster);
		if(sessionList==null){
			sessionList=(new ArrayList<RPCSession>());
			sessionMap.put(serverInfo.cluster,sessionList);
		}
		sessionList.add(session);
		if(logger.isInfoEnabled()){
			logger.info("create rpc session:"+session);
		}
	}
	//
	public void start() throws Exception {
		if(logger.isInfoEnabled()){
			logger.info(getClass().getSimpleName()+" start");
		}
		threadPoolManager.start();
		networkService.start();
		//
		checkSessionConnected();
		threadPoolManager.scheduleAtFixedRate(this::checkSessionConnected,30,30,TimeUnit.SECONDS);
		if(logger.isInfoEnabled()){
			logger.info(dump());
		}
	}
	//
	public void stop() throws Exception {
		if(logger.isInfoEnabled()){
			logger.info(getClass().getSimpleName()+" stop");
		}
		networkService.stop();
	}
	//
	public void addRPCServerInfo(RPCServerInfo info){
		List<RPCServerInfo>serverList=serverInfoMap.get(info.cluster);
		if(serverList==null){
			serverList=new ArrayList<>();
			serverInfoMap.put(info.cluster,serverList);
		}
		serverList.add(info);
	}
	//
	/**
	 * @return the networkService
	 */
	public NetworkService getNetworkService() {
		return networkService;
	}
	/**
	 * @param networkService the networkService to set
	 */
	public void setNetworkService(NetworkService networkService) {
		this.networkService = networkService;
	}
	/**
	 * @return the performMonitor
	 */
	public PerformMonitor getPerformMonitor() {
		return performMonitor;
	}
	/**
	 * @param performMonitor the performMonitor to set
	 */
	public void setPerformMonitor(PerformMonitor performMonitor) {
		this.performMonitor = performMonitor;
	}
	/**
	 * @return the topicMap
	 */
	public Map<String, Set<String>> getTopicMap() {
		return topicMap;
	}
	/**
	 * @param topicMap the topicMap to set
	 */
	public void setTopicMap(Map<String, Set<String>> topicMap) {
		this.topicMap = topicMap;
	}
	//
	//
	public void subscribe(String cluster,String topic){
		Set<String>topics=topicMap.get(cluster);
		if(topics==null){
			topics=new TreeSet<String>();
			topicMap.put(cluster, topics);
		}
		topics.add(topic);
	}
	/**
	 * @return the codecType
	 */
	public static String getCodecType() {
		return codecType;
	}
	/**
	 * @param codecType the codecType to set
	 */
	public static void setCodecType(String codecType) {
		RPCClient.codecType = codecType;
	}
	/**
	 * @return the uuid
	 */
	public String getUuid() {
		return uuid;
	}
	/**
	 * @param uuid the uuid to set
	 */
	public void setUuid(String uuid) {
		this.uuid = uuid;
	}
	/**
	 * @return the rpcTimeout
	 */
	public int getRpcTimeout() {
		return rpcTimeout;
	}
	/**
	 * @param rpcTimeout the rpcTimeout to set
	 */
	public void setRpcTimeout(int rpcTimeout) {
		this.rpcTimeout = rpcTimeout;
	}
	/**
	 * @return the syncLockMap
	 */
	public Map<Long, ResponseLock> getSyncLockMap() {
		return syncLockMap;
	}
	/**
	 * @param syncLockMap the syncLockMap to set
	 */
	public void setSyncLockMap(Map<Long, ResponseLock> syncLockMap) {
		this.syncLockMap = syncLockMap;
	}
	/**
	 * @return the pushMessageCallback
	 */
	public RPCMessageCallback getPushMessageCallback() {
		return pushMessageCallback;
	}
	/**
	 * @param pushMessageCallback the pushMessageCallback to set
	 */
	public void setPushMessageCallback(RPCMessageCallback pushMessageCallback) {
		this.pushMessageCallback = pushMessageCallback;
	}
	/**
	 * @return the threadPoolManager
	 */
	public ThreadPoolManager getThreadPoolManager() {
		return threadPoolManager;
	}
	/**
	 * @param threadPoolManager the threadPoolManager to set
	 */
	public void setThreadPoolManager(ThreadPoolManager threadPoolManager) {
		this.threadPoolManager = threadPoolManager;
	}
	//
	private void checkSessionConnected(){
		sessionMap.forEach((cluster,sessionList)->{
			sessionList.forEach(session->{
				synchronized (session) {
					if(!session.isConnected()){
						if(logger.isWarnEnabled()){
							logger.warn("rpc session "+session+" is not connected");
						}
						networkService.connectToServer(session);
					}
				}
			});
		});
	}
	
	//
	public <T> T createSyncRemote(Class<T> interfaceClass,String cluster){
		return createService(
				interfaceClass,
				cluster,
				syncServiceMap,
				syncInvocationHandlerMap,
				false);
	}
	//
	public <T> T createAsyncRemote(Class<T> interfaceClass,String cluster){
		return createService(
				interfaceClass,
				cluster,
				asyncServiceMap,
				asyncInvocationHandlerMap,
				true);
	}
	//
	@SuppressWarnings("unchecked")
	private <T> T createService(
			Class<T> interfaceClass,
			String cluster,
			Map<String,Object> serviceMap,
			Map<String,RPCInvocationHandler> handlerMap,
			boolean isAsync){
		String name=cluster+"."+interfaceClass.getSimpleName();
		if(serviceMap.containsKey(name)){
			return (T) serviceMap.get(name);
		}
		checkClass(interfaceClass);
		//
		List<RPCSession>sessions=sessionMap.get(cluster);
		if(sessions==null){
			throw new IllegalArgumentException("can not find cluster with name:"+cluster);
		}
		//
		RPCInvocationHandler handler=handlerMap.get(cluster);
		if(handler==null){
			if(isAsync){
				handler=new RPCAsyncInvocationHandler(this,sessions);
			}else{
				handler=new RPCSyncInvocationHandler(this,sessions);
			}
			handlerMap.put(cluster,handler);
		}
		//
		Object obj=Proxy.newProxyInstance(
				interfaceClass.getClassLoader(),
				new Class[]{interfaceClass},
				handler);
		serviceMap.put(name,obj);
		return (T)obj;
	}

	//检查类不是是一个接口,同时检查方法里的参数是否都可以序列化
	public void checkClass(Class<?> interfaceClass){
		if(!interfaceClass.isInterface()){
			throw new IllegalArgumentException("registerService failed."+interfaceClass
					+" is not Interface."); 
		}
		for(Method m:interfaceClass.getDeclaredMethods()){
	        if(!Modifier.isPublic(m.getModifiers())){
	        	continue;
	        }
	        if(Modifier.isStatic(m.getModifiers())){
	        	continue;
	        }
	        for(Class<?> c:m.getParameterTypes()){
				if (c.isPrimitive()) {
					continue;
				}
				if(Modifier.isAbstract(c.getModifiers())||
						Modifier.isInterface(c.getModifiers())){
					continue;
				}
	        }
		}
	}
	/**
	 * @return the contextClassLoader
	 */
	public ClassLoader getContextClassLoader() {
		return contextClassLoader;
	}
	/**
	 * @param contextClassLoader the contextClassLoader to set
	 */
	public void setContextClassLoader(ClassLoader contextClassLoader) {
		this.contextClassLoader = contextClassLoader;
	}
	//
	public String dump() {
		ObjectPrinter op=new ObjectPrinter();
		op.section(getClass().getSimpleName());
		op.print("uuid",uuid);
		op.print("codecType",codecType);
		op.print("rpcTimeout",rpcTimeout);
		int index=1;
		op.secondSection("serverList");
		for(List<RPCServerInfo> e:serverInfoMap.values()){	
			for (RPCServerInfo info : e) {
				op.print(""+index,info);
				index++;
			}
		}
		op.secondSection("topics");
		topicMap.forEach((k,v)->{
			op.print(k,v.toString());
		});
		return op.toString();
	}
}
